package com.weimob.mcloud.wechatwork.customer.job;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
import com.google.common.collect.Lists;
import com.weimob.elasticjob.annotation.ElasticJob;
import com.weimob.elasticjob.executor.context.ExecutorContext;
import com.weimob.marketing.syswechatworkusermanagement.wechatworkusermanagement.service.user.WechatWorkEmployeeExport;
import com.weimob.mcloud.wechatwork.core.adapter.enums.cdp.CDPMarkTypeEnum;
import com.weimob.mcloud.wechatwork.core.adapter.enums.cdp.CdpMarkMessageEventEnum;
import com.weimob.mcloud.wechatwork.core.adapter.pojo.dto.cdp.request.customer.CDPUserMarkDTO;
import com.weimob.mcloud.wechatwork.core.adapter.service.cdp.CdpUserAdapterService;
import com.weimob.mcloud.wechatwork.core.base.SingleRequest;
import com.weimob.mcloud.wechatwork.core.entity.privatesphere.PrivateSphere;
import com.weimob.mcloud.wechatwork.core.entity.relation.CustomerEmployeeRelation;
import com.weimob.mcloud.wechatwork.core.entity.relation.customer.Customer;
import com.weimob.mcloud.wechatwork.core.enums.customer.CustomerStatusEnum;
import com.weimob.mcloud.wechatwork.core.enums.customer.ExternalUserTypeEnum;
import com.weimob.mcloud.wechatwork.core.service.customer.CustomerService;
import com.weimob.mcloud.wechatwork.core.service.customer.request.CustomerScrollRequest;
import com.weimob.mcloud.wechatwork.core.service.customer.response.CustomerScrollResponse;
import com.weimob.mcloud.wechatwork.core.service.privatesphere.PrivateSphereService;
import com.weimob.mcloud.wechatwork.customer.constant.Constants;
import com.weimob.mcloud.wechatwork.customer.domain.CdpDataJsonInfo;
import com.weimob.mcloud.wechatwork.customer.mq.handler.externaluser.AddCorpExternaluserSecondHandler;
import com.weimob.mcloud.wechatwork.customer.service.user.enums.CDPFlowStatusEnum;
import com.weimob.mcloud.wechatwork.customer.third.core.CustomerServiceProxy;
import com.weimob.mcloud.wechatwork.customer.third.customer.CustomerThird;
import com.weimob.soa.common.response.SoaResponse;
import com.weimob.soa.common.utils.SoaUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StopWatch;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.CompletableFuture;

/**
 * Create Time: 2022年05月25日 22:14
 *
 * @author maosheng.li
 **/
@Slf4j
@Component
public class CustomerJob {

    @Reference
    private CustomerService customerService;

    @Autowired
    private CustomerServiceProxy customerServiceProxy;

    @Reference
    private WechatWorkEmployeeExport wechatWorkEmployeeExport;


    @Reference(timeout = 20000)
    PrivateSphereService privateSphereService;

    @Autowired
    private CustomerThird customerThird;

    @Autowired
    private AddCorpExternaluserSecondHandler addCorpExternaluserSecondHandler;

    @Resource(name = "customerFlushSearchThreadPool")
    ThreadPoolTaskExecutor customerFlushSearchThreadPool;

    @Resource(name = "customerFlushSearchThreadPoolNested")
    ThreadPoolTaskExecutor customerFlushSearchThreadPoolNested;


    @Value("${flush.customer.search.partition.count:100}")
    private Integer partitionCount;

    @Value("${scroll.pagesize:20}")
    private Integer scrollPageSize;

    @Value("${scroll.partition.count:30}")
    private Integer scrollPartitionCount;

    @Value("${scroll.retry.count:2}")
    private Integer scrollRetryCount;

    @ApolloJsonValue("${customer.flush.corpId:[]}")
    private List<String> customerFlushCorpId;

    /**
     * 企微助手产品id
     */
    @Value("${aewc.productId:203}")
    private Long aewcProductId;

    @Reference
    CdpUserAdapterService cdpUserAdapterService;

    @Value("${transfer.page.size:500}")
    private Integer transferPageSize;

    @Value("${customerFirstAddTimeFlushFlag:true}")
    private Boolean customerFirstAddTimeFlushFlag;


    @ElasticJob(name = "flushCustomerToSearch", cron = "0 0 0 30 12 ? 2099")
    public void flushCustomerToSearch() {
        log.info("flushCustomerToSearch start");
        Long start = System.currentTimeMillis();

        // apollo 没有指定就刷全量店铺
        List<PrivateSphere> flushList = Lists.newArrayList();
        if (!CollectionUtils.isEmpty(customerFlushCorpId)) {
            for (String corpId : customerFlushCorpId) {
                PrivateSphere privateSphere = new PrivateSphere();
                privateSphere.setCorpId(corpId);
                SoaResponse<PrivateSphere, Void> soaResponse = privateSphereService.simpleFetch(privateSphere);
                if (soaResponse != null && soaResponse.getResponseVo() != null) {
                    flushList.add(soaResponse.getResponseVo());
                }
            }
        } else {
            PrivateSphere query = new PrivateSphere();
            query.setStatus(1);
            query.setProductId(aewcProductId);
            flushList = SoaUtil.unpack(privateSphereService.search(query));
        }
        log.info("flushCustomerStatus,flushList:{}", JSON.toJSONString(flushList));

        if (CollectionUtils.isEmpty(flushList)) {
            log.info("待刷数据为空，请检查");
            return;
        }

        //默认并发100个corpId跑
        List<List<PrivateSphere>> partitionList = Lists.partition(flushList, partitionCount);
        for (List<PrivateSphere> privateSphereList : partitionList) {
            CompletableFuture<Object>[] completableFutures = new CompletableFuture[privateSphereList.size()];
            for (int i = 0; i < privateSphereList.size(); i++) {
                PrivateSphere privateSphereResult = privateSphereList.get(i);
                completableFutures[i] = CompletableFuture.supplyAsync(() -> {
                    log.info("flushCustomerToSearch,privateSphereResult:{}, start", JSON.toJSONString(privateSphereResult));
                    scrollCustomerToES(privateSphereResult);
                    log.info("flushCustomerToSearch,privateSphereResult:{}, end", JSON.toJSONString(privateSphereResult));
                    return null;
                }, customerFlushSearchThreadPool);
            }
            //等待执行完成
            CompletableFuture.allOf(completableFutures).join();
        }

        log.info("flushCustomerToSearch end ,cost {}", System.currentTimeMillis() - start);
    }

    /**
     * 滚动数据到ES
     *
     * @param privateSphereResult
     */
    private void scrollCustomerToES(PrivateSphere privateSphereResult) {
        Integer retryCount = 0;
        Integer nextTableNum = 0;
        Long nextId = 0L;
        Boolean hashNext = true;
        CustomerScrollRequest customerScrollRequest = new CustomerScrollRequest();
        do {
            customerScrollRequest.setCorpid(privateSphereResult.getCorpId());
            customerScrollRequest.setNextTableNum(nextTableNum);
            customerScrollRequest.setNextId(nextId);
            customerScrollRequest.setPageSize(scrollPageSize);
            log.info("scrollCustomerToES, customerScrollRequest:{}", JSON.toJSONString(customerScrollRequest));
            SoaResponse<CustomerScrollResponse, ?> customerScrollResponseSoaResponse = customerService.scrollCustomer(customerScrollRequest);
            CustomerScrollResponse responseVo = customerScrollResponseSoaResponse.getResponseVo();
            if (responseVo == null) {
                //睡一秒重试一下吧（大客跑一半就凉了）
                try {
                    Thread.sleep(500);
                    retryCount++;
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("scrollCustomerToES error:", e);
                }
                log.error("responseVo is null , privateSphere:{}, nextTableNum:{}, nextId:{}", JSON.toJSONString(privateSphereResult), nextTableNum, nextId);
                if (retryCount > scrollRetryCount) {
                    //大于重试次数 return
                    return;
                } else {
                    continue;
                }
            }
            hashNext = responseVo.getHasNext();
            nextId = responseVo.getNextId();
            nextTableNum = responseVo.getNextTableNum();
            //开始处理数据
            if (!CollectionUtils.isEmpty(responseVo.getDataList())) {
                List<List<Customer>> partition = Lists.partition(responseVo.getDataList(), scrollPartitionCount);
                for (List<Customer> customerList : partition) {
                    CompletableFuture<Object>[] completableFutures = new CompletableFuture[customerList.size()];
                    for (int i = 0; i < customerList.size(); i++) {
                        Customer customer = customerList.get(i);
                        completableFutures[i] = CompletableFuture.supplyAsync(() -> {
                            Long superWid = customerThird.getSuperWid(privateSphereResult.getBosId(), customer.getExternalUserid(), privateSphereResult.getCorpId());
                            if (superWid != null) {
                                if (ExternalUserTypeEnum.WECHAT_CUSTOMER.equals(customer.getExternalUserType())) {
                                    //外部联系人
                                    addCorpExternaluserSecondHandler.updateCustomerSearchHisMsg(superWid, privateSphereResult.getBosId(),
                                            privateSphereResult.getProductInstanceId(), privateSphereResult.getProductId());
                                }
                            }
                            return null;
                        }, customerFlushSearchThreadPoolNested);
                    }
                    //等待执行完成
                    CompletableFuture.allOf(completableFutures).join();
                }
            }
        } while (hashNext);
    }

    /**
     * 获取index
     *
     * @param shardingTotal 总数
     * @param shardingIndex 当前index
     * @param endIndex      总index
     * @return
     */
    public int[] getIndex(Integer shardingTotal, Integer shardingIndex, int endIndex) {
        int size = (int) Math.ceil((double) endIndex / shardingTotal);
        int start = shardingIndex * size;
        int end = Math.min(start + size, endIndex);
        return new int[]{start, end};
    }


    @ElasticJob(name = "customerInfoTransfer")
    public void customerInfoTransfer(ExecutorContext executorContext) {
        Integer shardingTotal = executorContext.getShardingTotal();
        Integer shardingIndex = executorContext.getShardingIndex();
        int[] index = getIndex(shardingTotal, shardingIndex, 64);
        int startIndex = index[0];
        int endIndex = index[1];
        log.info("customerInfoTransfer start,start:{},end:{},param:{}", startIndex, endIndex, JSON.toJSON(executorContext));

        StopWatch stopWatch = new StopWatch();
        stopWatch.start("customerInfoTransfer");

        SingleRequest<Customer> request = new SingleRequest<>();
        Map<String, Object> extInfo = new HashMap<>(1);
        request.setPageSize(transferPageSize);
        int total = 0;
        Map<String, PrivateSphere> hashMap = new HashMap<>(64);
        //分批查询tableNum
        for (int i = startIndex; i < endIndex; i++) {
            extInfo.put("index", i);
            List<Customer> list;
            String id = null;
            int tableTotal = 0;
            do {
                Customer customerReq = new Customer();
                customerReq.setStatus(CustomerStatusEnum.UNAUTHORIZED);
                customerReq.setId(id);
                request.setExtInfo(extInfo);
                request.setData(customerReq);
                list = SoaUtil.unpack(customerService.fetchCustomersByIndex(request));
                if (CollectionUtils.isEmpty(list)) {
                    //查询下一个表
                    log.info("fetchCustomersByIndex table is complete,current index:{},tableTotal:{}", i, tableTotal);
                    break;
                }
                tableTotal += list.size();
                total += list.size();
                //处理
                for (Customer item : list) {
                    //游标-id赋值
                    id = item.getId();
                    // 获取私域
                    String corpId = item.getCorpId();
                    PrivateSphere responseVo = getPrivateSphere(hashMap, corpId);
                    if (Objects.isNull(responseVo)) {
                        log.warn("customerInfoTransfer privateSphere#simpleFetch is null,corpId:{}", corpId);
                        continue;
                    }
                    //私域
                    item.setPrivateSphere(responseVo);
                    //获取wid
                    Long superWid = customerThird.getSuperWid(responseVo.getBosId(), item.getExternalUserid(), corpId);

                    if (Objects.isNull(superWid)) {
                        log.warn("superWid is null");
                        continue;
                    }
                    //转化成消息对象
                    CDPUserMarkDTO cdpUserMarkDTO = convertCDPMsg(item, superWid);
                    //发送
                    log.info("cdp customerInfoTransfer,index:{},request {}", i, JSON.toJSONString(cdpUserMarkDTO));
                    cdpUserAdapterService.markCDPUserMsg(cdpUserMarkDTO);
                }
            } while (!CollectionUtils.isEmpty(list));

        }

        stopWatch.stop();
        log.info("customerInfoTransfer is complete,initIndex:{},endIndex:{},total:{},耗时：{} seconds", startIndex, endIndex, total, stopWatch.getTotalTimeSeconds());

    }

    /**
     * 对象转换
     *
     * @param customer customer
     * @return CDPUserMarkDTO
     */
    private CDPUserMarkDTO convertCDPMsg(Customer customer, Long superWid) {
        PrivateSphere privateSphere = customer.getPrivateSphere();
        CDPUserMarkDTO cdpUserMarkDTO = new CDPUserMarkDTO();
        cdpUserMarkDTO.setuKey(String.valueOf(superWid));
        cdpUserMarkDTO.setuKeyType("wid");
        cdpUserMarkDTO.setsChannelType(203);
        cdpUserMarkDTO.setsChannelId(privateSphere.getProductInstanceId());
        cdpUserMarkDTO.setBosId(privateSphere.getBosId());
        cdpUserMarkDTO.setUpdateTime(String.valueOf(System.currentTimeMillis()));
        //jsonInfo
        CdpDataJsonInfo cdpDataJsonInfo = new CdpDataJsonInfo();
        cdpDataJsonInfo.setProductInstanceId(privateSphere.getProductInstanceId());
        cdpDataJsonInfo.setWid(superWid);
        cdpDataJsonInfo.setCorpId(privateSphere.getCorpId());
        cdpDataJsonInfo.setExternalUserId(customer.getExternalUserid());
        cdpDataJsonInfo.setCustomerStatus(customer.getStatus().getCode());


        cdpUserMarkDTO.setDataJson(JSON.parseObject(JSON.toJSONString(cdpDataJsonInfo)));
        cdpUserMarkDTO.setCdpMarkTypeEnum(CDPMarkTypeEnum.CDP_USER_ATTRIBUTE);
        cdpUserMarkDTO.setCdpMarkMessageEventEnum(CdpMarkMessageEventEnum.CUSTOMER_ATTRIBUTE);

        return cdpUserMarkDTO;
    }


    @ElasticJob(name = "loseStatusTransfer")
    public void loseStatusTransfer(ExecutorContext executorContext) {
        Integer shardingTotal = executorContext.getShardingTotal();
        Integer shardingIndex = executorContext.getShardingIndex();
        Map<String, String> triggerCustomParameter = executorContext.getTriggerCustomParameter();
        log.info("loseStatusTransfer.triggerCustomParameter:{}", JSON.toJSONString(triggerCustomParameter));
        String corpId = triggerCustomParameter.get("corpId");
        String diyStartIndex = triggerCustomParameter.get("startIndex");
        String diyEndIndex = triggerCustomParameter.get("endIndex");
        if (!StringUtils.hasText(corpId)) {
            log.error("loseStatusTransfer.triggerCustomParameter corpId is null");
            return;
        }
        int[] index = getIndex(shardingTotal, shardingIndex, 64);
        int startIndex = index[0];
        int endIndex = index[1];
        //自定义startIndex，endIndex
        if (StringUtils.hasText(diyStartIndex) && StringUtils.hasText(diyEndIndex)) {
            startIndex = Integer.parseInt(diyStartIndex);
            endIndex = Integer.parseInt(diyEndIndex);
        }
        log.info("loseStatusTransfer start,start:{},end:{},param:{}", startIndex, endIndex, JSON.toJSON(executorContext));

        StopWatch stopWatch = new StopWatch();
        stopWatch.start("loseStatusTransfer");

        int total = 0;
        Map<String, PrivateSphere> hashMap = new HashMap<>(64);
        //分批查询tableNum
        for (int i = startIndex; i < endIndex; i++) {

            String id = null;
            List<Customer> list;
            int tableTotal = 0;
            do {
                //查询
                list = selectListByIndex(id, i, corpId);
                if (CollectionUtils.isEmpty(list)) {
                    //查询下一个表
                    log.info("fetchCustomersByIndex table is complete,current index:{},tableTotal:{}", i, tableTotal);
                    break;
                }
                tableTotal += list.size();
                total += list.size();
                //处理
                for (Customer item : list) {
                    //游标-id赋值
                    id = item.getId();
                    //status判断
                    CustomerStatusEnum statusEnum = item.getStatus();
                    if (Objects.isNull(statusEnum)) {
                        log.warn("loseStatusTransfer.statusEnum is null");
                        continue;
                    }
                    // 获取私域
                    PrivateSphere privateSphere = getPrivateSphere(hashMap, corpId);
                    if (Objects.isNull(privateSphere)) {
                        log.warn("loseStatusTransfer privateSphere#simpleFetch is null,corpId:{}", corpId);
                        continue;
                    }
                    //获取wid
                    Long superWid = customerThird.getSuperWid(privateSphere.getBosId(), item.getExternalUserid(), corpId);

                    if (Objects.isNull(superWid)) {
                        log.warn("superWid is null,externalUserid is:{}", item.getExternalUserid());
                        continue;
                    }
                    Long count = null;
                    //customerStatus 这个字段必须要有，flowStatus 只有是有效状态下必须要有
                    if (CustomerStatusEnum.DEFAULT.equals(statusEnum)) {
                        count=customerServiceProxy.fetchCustomerEmployeeRelationCount(corpId,id,true,CustomerStatusEnum.DEFAULT);
                    }
                    //转化消息体
                    CDPUserMarkDTO cdpUserMarkDTO = convertDelEventBody(item, superWid, count, privateSphere);
                    cdpUserAdapterService.markCDPUserMsg(cdpUserMarkDTO);
                }
            } while (!CollectionUtils.isEmpty(list));

        }

        stopWatch.stop();
        log.info("loseStatusTransfer is complete,corpId:{},initIndex:{},endIndex:{},total:{},耗时：{} seconds", corpId, startIndex, endIndex, total, stopWatch.getTotalTimeSeconds());

    }

    private CDPUserMarkDTO convertDelEventBody(Customer customer, Long superWid, Long count, PrivateSphere privateSphere) {
        CDPUserMarkDTO cdpUserMarkDTO = new CDPUserMarkDTO();
        cdpUserMarkDTO.setuKey(String.valueOf(superWid));
        cdpUserMarkDTO.setuKeyType("wid");
        cdpUserMarkDTO.setsChannelType(203);
        cdpUserMarkDTO.setsChannelId(privateSphere.getProductInstanceId());
        cdpUserMarkDTO.setBosId(privateSphere.getBosId());
        //2022-08-01 00:00:00 时间戳
        cdpUserMarkDTO.setUpdateTime(Constants.COMPENSATE_UPDATE_TIME);
        //dataJson
        CdpDataJsonInfo cdpDataJsonInfo = new CdpDataJsonInfo();
        cdpDataJsonInfo.setProductInstanceId(privateSphere.getProductInstanceId());
        cdpDataJsonInfo.setWid(superWid);
        cdpDataJsonInfo.setCorpId(privateSphere.getCorpId());
        cdpDataJsonInfo.setExternalUserId(customer.getExternalUserid());
        cdpDataJsonInfo.setWechatNickName(customer.getName());

        CustomerStatusEnum customerStatusEnum = customer.getStatus();
        cdpDataJsonInfo.setCustomerStatus(customerStatusEnum.getCode());
        if (Objects.nonNull(count)) {
            if (count > 0) {
                cdpDataJsonInfo.setFlowStatus(CDPFlowStatusEnum.NOT_FLOW_STATUS.getCode());
            } else {
                cdpDataJsonInfo.setFlowStatus(CDPFlowStatusEnum.LOST_STATUS.getCode());
            }
        }
        cdpUserMarkDTO.setDataJson(JSON.parseObject(JSON.toJSONString(cdpDataJsonInfo)));
        cdpUserMarkDTO.setCdpMarkTypeEnum(CDPMarkTypeEnum.CDP_USER_ATTRIBUTE);
        cdpUserMarkDTO.setCdpMarkMessageEventEnum(CdpMarkMessageEventEnum.CUSTOMER_ATTRIBUTE);
        return cdpUserMarkDTO;
    }


    private List<Customer> selectListByIndex(String id, int index, String corpId) {
        Customer req = new Customer();
        req.setCorpId(corpId);
        req.setId(id);
        Map<String, Object> extInfo = new HashMap<>(1);
        extInfo.put("index", index);
        SingleRequest<Customer> request = new SingleRequest<>();
        request.setPageSize(transferPageSize);
        request.setExtInfo(extInfo);
        request.setData(req);
        return SoaUtil.unpack(customerService.fetchCustomersByIndex(request));
    }

    /**
     * 查询私域
     *
     * @param hashMap
     * @param corpId
     * @return
     */
    private PrivateSphere getPrivateSphere(Map<String, PrivateSphere> hashMap, String corpId) {
        // 获取私域
        if (!hashMap.containsKey(corpId)) {
            PrivateSphere privateSphere = new PrivateSphere();
            privateSphere.setCorpId(corpId);
            SoaResponse<PrivateSphere, Void> soaResponse = privateSphereService.simpleFetch(privateSphere);
            PrivateSphere result = soaResponse.getResponseVo();
            //本地缓存
            hashMap.put(corpId, result);
        }
        return hashMap.get(corpId);
    }




    @ElasticJob(name = "customerFirstAddTimeFlush")
    public void customerFirstAddTimeFlush(ExecutorContext executorContext) {
        Integer shardingTotal = executorContext.getShardingTotal();
        Integer shardingIndex = executorContext.getShardingIndex();
        int[] index = getIndex(shardingTotal, shardingIndex, 64);
        int startIndex = index[0];
        int endIndex = index[1];
        Map<String, String> triggerCustomParameter = executorContext.getTriggerCustomParameter();
        log.info("customerFirstAddTimeFlush.triggerCustomParameter:{}", JSON.toJSONString(triggerCustomParameter));
        String targetCorpId = triggerCustomParameter.get("corpId");
        String diyStartIndex = triggerCustomParameter.get("startIndex");
        String diyEndIndex = triggerCustomParameter.get("endIndex");

        //自定义startIndex，endIndex
        if (StringUtils.hasText(diyStartIndex) && StringUtils.hasText(diyEndIndex)) {
            startIndex = Integer.parseInt(diyStartIndex);
            endIndex = Integer.parseInt(diyEndIndex);
        }

        log.info("customerFirstAddTimeFlush start,start:{},end:{},param:{}", startIndex, endIndex, JSON.toJSON(executorContext));

        StopWatch stopWatch = new StopWatch();
        stopWatch.start("customerFirstAddTimeFlush");

        SingleRequest<Customer> request = new SingleRequest<>();
        Map<String, Object> extInfo = new HashMap<>(1);
        request.setPageSize(transferPageSize);
        int total = 0;
        Map<String, PrivateSphere> hashMap = new HashMap<>(64);
        //分批查询tableNum
        for (int i = startIndex; i < endIndex; i++) {
            extInfo.put("index", i);
            List<Customer> list;
            String id = null;
            int tableTotal = 0;
            do {
                list = selectListByIndex(id, i, targetCorpId);
                if (CollectionUtils.isEmpty(list)) {
                    //查询下一个表
                    log.info("fetchCustomersByIndex table is complete,current index:{},tableTotal:{}", i, tableTotal);
                    break;
                }
                tableTotal += list.size();
                total += list.size();
                //处理
                for (Customer item : list) {
                    //游标-id赋值
                    id = item.getId();

                    if (!customerFirstAddTimeFlushFlag) {
                        log.info("客户首次添加事件刷新停止,current index:{},id:{}", i, id);
                        return;
                    }

                    // addtime不为空 过滤 不用刷
                    if (item.getFirstAddTime() != null) {
                        continue;
                    }

                    // 获取私域
                    String corpId = item.getCorpId();
                    PrivateSphere responseVo = getPrivateSphere(hashMap, corpId);
                    if (Objects.isNull(responseVo)) {
                        log.warn("customerFirstAddTimeFlush privateSphere#simpleFetch is null,corpId:{}", corpId);
                        continue;
                    }
                    //私域
                    item.setPrivateSphere(responseVo);


                    // 查关系 更新首次添加事件
                    List<CustomerEmployeeRelation> relationList = null;
                    try {
                        relationList = customerServiceProxy.fetchCustomerEmployeeRelation(item.getId(), null, responseVo, null, null, null);
                    } catch (Exception e) {
                        log.error("customerFirstAddTimeFlush fetchCustomerEmployeeRelation error,id:{},corpId:{}", item.getId(), corpId, e);
                        continue;
                    }

                    //NOSONAR
                    if(!CollectionUtils.isEmpty(relationList)){
                        Date firstAddTime = relationList.get(0).getAddTime();
                        if (relationList.size() > 1) {
                            for (CustomerEmployeeRelation employeeRelation : relationList) {
                                if (firstAddTime.getTime() > employeeRelation.getAddTime().getTime()) {
                                    firstAddTime = employeeRelation.getAddTime();
                                }
                            }
                        }


                        // 更新
                        Customer updateCustomer = new Customer();
                        updateCustomer.setId(item.getId());
                        updateCustomer.setFirstAddTime(firstAddTime);
                        customerService.updateCustomer(updateCustomer);
                    }


                    //获取wid
                    try {
                        // 更新ES
                        if(ExternalUserTypeEnum.WECHAT_CUSTOMER.getCode().equals(item.getExternalUserType().getCode())) {
                            Long superWid = customerThird.getSuperWid(responseVo.getBosId(), item.getExternalUserid(), corpId);
                            if (Objects.isNull(superWid)) {
                                log.warn("flushAddTime superWid is null");
                                continue;
                            }

                            addCorpExternaluserSecondHandler.updateCustomerSearchHisMsg(superWid, responseVo.getBosId(),
                                    responseVo.getProductInstanceId(), responseVo.getProductId());
                        }
                    } catch (Exception e) {
                        log.error("customerFirstAddTimeFlush同步ES失败, externalUserId:{}", item.getExternalUserid(), e);
                    }

                }
            } while (!CollectionUtils.isEmpty(list));

        }

        stopWatch.stop();
        log.info("customerFirstAddTimeFlush is complete,initIndex:{},endIndex:{},total:{},耗时：{} seconds", startIndex, endIndex, total, stopWatch.getTotalTimeSeconds());

    }

}
